#include "AmqpWorker.h"
#include <iostream>

using namespace AmqpClient;

#define QUEUE_MANAGER "web.server.receipt"
#define QUEUE_TASKS   "task.%d"
#define RK_BROCAST    "web.server.brocast"

bool AmqpWorker::Connect(const std::string& host, int port, const std::string& username, const std::string& password)
{
	bool ok = false;
	try {
	//	channel = Channel::Create(host, port, username, password);
		Channel::OpenOpts login;
		login.host = host;
		login.port = port;
		login.auth = Channel::OpenOpts::BasicAuth(username, password);
		channel = Channel::Open(login);
		ok = init_connection();
	} catch(std::exception& e) {
		std::cerr << "Connect failed: " << e.what() << std::endl;
	}
	if (!ok) {
		if (channel) {
			channel = nullptr;
			commandQueue = "";
		}
		return false;
	}
	return true;
}

void AmqpWorker::Register()
{
	if (withManager) {
		json j_message{
			{"command", "Register"}, {"queue", commandQueue}, {"taskType", taskType}
		};
		send_message(j_message);
	}
}

void AmqpWorker::Unregister()
{
	if (withManager) {
		json j_message{
			{"command", "Unregister"}, {"workerId", workerId}
		};
		send_message(j_message);
	}
}

bool AmqpWorker::init_connection()
{
	char task_queue[256];
	sprintf(task_queue, QUEUE_TASKS, taskType);
	taskQueue = task_queue;
	
	if (withManager) {
		channel->DeclareQueue(QUEUE_MANAGER, /*passive=*/true);
	}
	channel->DeclareQueue(taskQueue, /*passive=*/true);	
	
	if (withManager) {
		commandQueue = channel->DeclareQueue("", /*passive=*/false, /*durable=*/false, /*exclusive=*/true, /*auto_delete=*/true);
		channel->BindQueue(commandQueue, /*exchange_name=*/"amq.direct", /*routing_key=*/RK_BROCAST);
		channel->BasicConsume(commandQueue, "", /*no_local=*/true, /*no_ack=*/false, /*exclusive=*/true);
	} else {
		taskConsumerTag = channel->BasicConsume(taskQueue, /*consumer_tag=*/"",
				/*no_local=*/true, /*no_ack=*/false, /*exclusive=*/false, /*message_prefetch_count=*/50);
	}
	return true;
}

void AmqpWorker::send_message(const json& j_message)
{
	std::string j_string = j_message.dump();
	BasicMessage::ptr_t message = BasicMessage::Create(j_string);
	channel->BasicPublish(/*exchange_name=*/"", /*routing_key=*/QUEUE_MANAGER, message);
	
	std::cout << "send: " << j_string << std::endl;
}

bool AmqpWorker::recv_message(json& j_message, Envelope::DeliveryInfo& delivery_info, int timeout)
{
	Envelope::ptr_t message;
	if (!channel->BasicConsumeMessage(message, timeout)) {
		return false;
	}
	
	delivery_info = message->GetDeliveryInfo();
	std::string j_string = message->Message()->Body();
	j_message = json::parse(j_string);
	std::cout << "recv: " << j_string << std::endl;
	return true;
}

void AmqpWorker::drain_events(int timeout)
{
	while (!complete_tasks.empty()) {
		TaskInfo task = complete_tasks.front();
		complete_tasks.pop();
		endTask(task.taskId);
		channel->BasicAck(task.deliveryInfo);
	}
	
	json j_message;
	Envelope::DeliveryInfo deliveryInfo;
	if (!recv_message(j_message, deliveryInfo, timeout)) {
		return;
	}
	
	std::string cmd = j_message["command"];
	if (cmd == "Task") {
		TaskInfo task{j_message["taskId"], j_message["url"], deliveryInfo};
		onTask(task);
	} else if (withManager) {
		if (cmd == "Register.OK") {
			onRegisterOK(j_message["workerId"]);
		} else if (cmd == "Register.Again") {
			onStop();
			Register();
		} else if (cmd == "Status") {
			onStatus(j_message["statusType"]);
		} else if (cmd == "Stop") {
			onStop();
		}
		channel->BasicAck(deliveryInfo);
	}
}

void AmqpWorker::onRegisterOK(const std::string& worker_id)
{
	workerId = worker_id;
	taskConsumerTag = channel->BasicConsume(taskQueue, /*consumer_tag=*/"",
			/*no_local=*/true, /*no_ack=*/false, /*exclusive=*/false, /*message_prefetch_count=*/50);
}

void AmqpWorker::onStatus(int statusType)
{
	json j_message{
		{"command", "Status"}, {"statusType", statusType}, {"workerId", workerId}, {"status", "ok"}
	};
	send_message(j_message);
}

void AmqpWorker::onStop()
{
	if (!taskConsumerTag.empty()) {
		channel->BasicCancel(taskConsumerTag);
		taskConsumerTag = "";
	}
}

void AmqpWorker::onTask(const TaskInfo& task)
{
	waiting_tasks.push(task);
	startTask(task.taskId);
}

void AmqpWorker::startTask(const std::string& task_id)
{
	if (withManager) {
		json j_message{
			{"command", "Task.Start"}, {"taskId", task_id}, {"workerId", workerId}
		};
		send_message(j_message);
	}
}

void AmqpWorker::endTask(const std::string& task_id)
{
	if (withManager) {
		json j_message{
			{"command", "Task.OK"}, {"taskId", task_id}, {"workerId", workerId}
		};
		send_message(j_message);
	}
}

bool AmqpWorker::pull_task(AmqpWorker::TaskInfo& task)
{
	if (waiting_tasks.empty())
		return false;
	task = waiting_tasks.front();
	waiting_tasks.pop();
	return true;
}

void AmqpWorker::finish_task(const AmqpWorker::TaskInfo& task)
{
	complete_tasks.push(task);
}
